package com.chukun.flink.stream.window.join;

import com.chukun.flink.stream.window.join.bean.ClickBean;
import com.chukun.flink.stream.window.join.bean.PrepareData;
import com.chukun.flink.stream.window.join.bean.UserTrade;
import com.chukun.flink.stream.window.join.extractor.ClickBeanEventTimestampExtractor;
import com.chukun.flink.stream.window.join.extractor.TradeEventTimestampExtractor;
import com.chukun.flink.stream.window.join.watermarker.ClickBeanWatermarkerGenerator;
import com.chukun.flink.stream.window.join.watermarker.TradeWatermarkerGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

/**
 * @author chukun
 * @version 1.0.0
 * @description intervaljoin基本操作，
 * @createTime 2022年05月26日 23:49:00
 */
public class IntervalJoinOperator {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<ClickBean> clickStream = env.fromCollection(PrepareData.getClicksData());

        DataStreamSource<UserTrade> tradeStream = env.fromCollection(PrepareData.getTradeData());

        // 设置原始用户点击数据的水印与事件时间
        SingleOutputStreamOperator<ClickBean> eventTimeClickStream = clickStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new ClickBeanWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new ClickBeanEventTimestampExtractor())
        );

        // 设置原始用户交易数据的水印与事件时间
        SingleOutputStreamOperator<UserTrade> eventTimeTradeStream = tradeStream.assignTimestampsAndWatermarks(
                // 生成水印 watermarker
                WatermarkStrategy.forGenerator((ctx) -> new TradeWatermarkerGenerator())
                        // 抽取事件产生的时间
                        .withTimestampAssigner((ctx) -> new TradeEventTimestampExtractor())
        );

        // 根据用户名分组
        KeyedStream<ClickBean, String> keyedClickStream = eventTimeClickStream.keyBy(ClickBean::getUser);

        // 根据用户名分组
        KeyedStream<UserTrade, String> keyedTradeStream = eventTimeTradeStream.keyBy(UserTrade::getUserName);

        SingleOutputStreamOperator<String> intervalJoinedStream = keyedClickStream.intervalJoin(keyedTradeStream)
                // 设置下限时间间隔为 -30分钟， 设置上限时间时间间隔为 20分钟
                .between(Time.minutes(-30), Time.minutes(20))
                // 连接的元素调用ProcessJoinFunction方法
                .process(new ProcessJoinFunction<ClickBean, UserTrade, String>() {
                    @Override
                    public void processElement(ClickBean clickBean, UserTrade userTrade, ProcessJoinFunction<ClickBean, UserTrade, String>.Context context, Collector<String> collector) throws Exception {
                        collector.collect(clickBean.toString() + " &&& " + userTrade.toString());
                    }
                });

        intervalJoinedStream.print("intervalJoin输出");

        env.execute("TumblingIntervalJoinOperator");
    }
}
